package com.atguigu.datastream.day07;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * ClassName: Flink05_Connect
 * Package: com.atguigu.day07
 * Description:
 *             连接（Connect）
 * @Author ChenJun
 * @Create 2023/4/14 18:15
 * @Version 1.0
 */
public class Flink05_Connect {
    public static void main(String[] args) throws Exception {

        //1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);


        //2. 创建两条流
        DataStreamSource<String> wordDStream = env.fromElements("a", "b", "c", "d");

        DataStreamSource<Integer> numberDStream = env.fromElements(1, 2, 3, 4);

        //TODO 3. 使用Union链接两条流
        ConnectedStreams<String, Integer> connect = wordDStream.connect(numberDStream);

        SingleOutputStreamOperator<String> result = connect.map(new CoMapFunction<String, Integer, String>() {
            @Override
            public String map1(String value) throws Exception {
                return value + "---> chenjun";
            }

            @Override
            public String map2(Integer value) throws Exception {
                return value * value + "";
            }
        });

        result.print();

        env.execute();

    }
}
